﻿using System;
using System.Threading;
using System.Collections.Generic;
using System.Linq;
using System.Runtime.Serialization;
using System.ServiceModel;
using System.Text;
using System.Net;
using Microsoft.WindowsAzure.ServiceRuntime;
using System.IO;


namespace HWCWorker_Worker
{
    using ElectionLogList = Dictionary<string, BinLogInfo>;    
    using ElectionSvcList = Dictionary<string, Func<MasterElectionServiceClient>>;
    using System.ServiceModel.Description;

    public class MasterSelectionEventArgs : EventArgs
    {
        public List<string> DaisyChain { get; internal set; }
        public string MasterId { get; internal set; }
    }

    [ServiceBehavior(InstanceContextMode = InstanceContextMode.Single)]
    public class MasterElectionService : IMasterElectionService
    {
        private const string MySQLEndpoint = "MySQL";
        private const string MySQLRoleName = "HWCWorker_Worker";
        private const int RetryCount = 6;
        private static TimeSpan TimeWait = TimeSpan.FromMinutes(6);

        private Dictionary<string, RoleInstanceEndpoint> _instanceIDToInstance;
        private string _endpointName;
        public RoleInstanceEndpoint _cachedMaster;
        private RoleInstance _thisInstance;
        private ElectionSvcList _electionServices;
        private ServiceHost _host;
        private ManualResetEvent _electionFinishedEvent;
        private bool _runningElection = false;
        private MySQLClient _sqlMonitor;
        private ElectionLogList _electionList = null;
        private int _serverCount;
        private int _serverId;
        private object _masterLock = new object();
                
        string instanceIdForLog = string.Empty;
        private LogClient _logger;

        /// <summary>
        /// Instantiates the service host for master election service
        /// </summary>
        /// <param name="endpointName"></param>
        public MasterElectionService(string endpointName)
        {
            _endpointName = endpointName;
            
            _electionFinishedEvent = new ManualResetEvent(false);
            _runningElection = false;
            _thisInstance = RoleEnvironment.CurrentRoleInstance;
            instanceIdForLog = _thisInstance.Id;
            _logger = new LogClient(instanceIdForLog);
            _instanceIDToInstance = new Dictionary<string, RoleInstanceEndpoint>();
            _electionServices = new Dictionary<string, Func<MasterElectionServiceClient>>();
            _sqlMonitor = new MySQLClient(MySQLEndpoint);

            foreach (var endpoint in RoleEnvironment.Roles[MySQLRoleName].Instances)
            {
                if (endpoint.Id != _thisInstance.Id)
                {
                    var uri = String.Format("net.tcp://{0}:{1}/MasterElectionService",
                                            endpoint.InstanceEndpoints[endpointName].IPEndpoint.Address,
                                            endpoint.InstanceEndpoints[endpointName].IPEndpoint.Port);

                    _electionServices.Add(endpoint.Id, () => new MasterElectionServiceClient(uri));
                }
                _instanceIDToInstance.Add(endpoint.Id, endpoint.InstanceEndpoints[endpointName]);
            }

            var hostUri = new Uri(String.Format("net.tcp://{0}:{1}/MasterElectionService",
                                                _thisInstance.InstanceEndpoints[endpointName].IPEndpoint.Address,
                                                _thisInstance.InstanceEndpoints[endpointName].IPEndpoint.Port));

            _host = new ServiceHost(this, hostUri);

            NetTcpBinding binding = new NetTcpBinding(SecurityMode.None, false);
            binding.MaxConnections = 300;
            binding.CloseTimeout = TimeSpan.FromHours(6);
            binding.OpenTimeout = TimeSpan.FromHours(6);
            binding.ReceiveTimeout = TimeSpan.FromHours(6);
            binding.SendTimeout = TimeSpan.FromHours(6);           
            _host.AddServiceEndpoint(typeof(IMasterElectionService), binding, hostUri);            
        }

        /// <summary>
        /// Opens the conenction to service host for the client to listen 
        /// </summary>
        public void StartListen()
        {
            while(_host.State != CommunicationState.Opened)
            {
                _host.Open(TimeSpan.FromSeconds(10));
            }
        }

        /// <summary>
        /// update master's endpoint 
        /// called during role start or whenever it's offline
        /// </summary>
        /// <returns>true if master endpoint is updated
        /// false if cached master is still running</returns>
        public bool RefreshMasterEndpoint()
        {
    
            determineServerID();

            if (ValidateMaster())
            {
                return false;
            }

            // master is not valid, start MasterElection process
            // (note: multiple nodes can start election simutaneously)
            try
            {
                _runningElection = true;
                StartElection();

                // declare victory!
                DeclareVictory();
            }
            finally
            {
                _runningElection = false;
            }
            return true;
        }

        public void getServerId()
        {
            determineServerID();
        }

        // return true if cached master is valid
        // return false if cached is updated for new master
        public bool ValidateMaster()
        {
            bool running = false;
            LogMsg("In Validate Master");
            try
            {
                // check against cached Master node first
                if (Master != null)
                {
                    LogMsg("ValidateMaster: Cached Master: {0}", Master.IPEndpoint.Address.ToString());

                    try
                    {
                        KeyValuePair<string, int> masterEpInfo = GetMySqlEndpointInfo(Master.RoleInstance.Id);
                        running = MySQLClient.IsMySqlRunning(masterEpInfo.Key, masterEpInfo.Value);
                        //_electionServices[Master.RoleInstance.Id]().IsRunning();
                        LogMsg("ValidateMaster: Cached Master IsRunning: {0}, {1}", Master.RoleInstance.Id, running.ToString());
                    }
                    catch (WebException)
                    {
                        LogMsg("Master offline:{0}", Master.RoleInstance.Id);
                        _cachedMaster = null;
                    }
                }
                else
                {
                    LogMsg("ValidateMaster: Cached Master: NULL");                    
                }
            }
            catch (Exception ex)
            {
                LogMsg("ValidateMaster:Error: {0}, {1}", ex.Message, ex.StackTrace);
            }
            LogMsg("End of ValidateMaster: Master Running: {0}", running.ToString());
            return running;
        }

        /// <summary>
        /// Determines the server id for a mysql instance
        /// </summary>
        private void determineServerID()
        {
            try
            {
                LogMsg("In determineServerID");
                var endpoints = RoleEnvironment.Roles[MySQLRoleName].Instances;
                _serverCount = endpoints.Count;
                var serverID = RoleEnvironment.CurrentRoleInstance.Id;
                //format example for serverID:
                // MySQL_WorkerRole_IN_1 (in cloud)
                // deployment(855).WorkerRole.1 (devFabric)
                int pos = serverID.LastIndexOf('.');
                if (pos == -1)
                {
                    pos = serverID.LastIndexOf('_');
                }
                serverID = serverID.Substring(pos + 1);
                int serverIndex = 0;
                if (!int.TryParse(serverID, out serverIndex))
                {
                    _serverId = -1;
                    return;
                }
                //RoleInstance Id is incanation number, which is zero based
                // MySQL server Id for replication cluster is one based
                _serverId = serverIndex + 1;
                LogMsg("DetermineServerId:Server Id: {0}", _serverId.ToString());                
            }
            catch (Exception ex)
            {
                LogMsg("determineServerId:Error: {0}", ex.StackTrace);
            }
        }

        /// <summary>
        /// Starts and proposes election for a master among the mysql instances
        /// </summary>
        private void StartElection()
        {
            try
            {
                LogMsg("In StartElection");
                _electionFinishedEvent.Reset();             
                // the host node will vote first, followed by asking other nodes to vote
                BinLogInfo logInfo = _sqlMonitor.GetBinLogInfo();
                CastElection(HostInstanceId, logInfo.LogId, logInfo.Position);
                List<Thread> proposers = new List<Thread>();
                foreach (string instanceID in _electionServices.Keys)
                {
                    // start proposing in parallel
                    Thread proposer = new Thread(ProposalThread);
                    proposer.Start(instanceID);
                    proposers.Add(proposer);
                }
                LogMsg("End of StartElection");
                // wait for all other nodes to cast vote
                _electionFinishedEvent.WaitOne(TimeSpan.FromMinutes(20));
            }
            catch (Exception ex)
            {
                LogMsg("StartElection:Error: {0}", ex.StackTrace);
            }
        }

        private void ProposalThread(object o)
        {
            string instanceID = (string)o;
            MasterElectionServiceClient client = _electionServices[instanceID]();
            
            int count = 0;
            while (count < RetryCount)
            {
                try
                {
                    // protect since peer instance may not be up at this moment.
                    client.ProposeElection(HostInstanceId);
                    break;
                }
                catch (Exception)
                {
                    count++;
                    Thread.Sleep(TimeWait);
                }
            }
            if (count == RetryCount)
            {
                LogMsg(" failed: ProposeElection to {0}", instanceID);
            }
        }


        /// <summary>
        /// Adds the bin log info for a mysql instance to election log list
        /// </summary>
        /// <param name="key"></param>
        /// <param name="logInfo"></param>
        private void AddElectionListEntry(string key, BinLogInfo logInfo)
        {
            lock (_masterLock)
            {
                if (_electionList == null)
                {
                    _electionList = new Dictionary<string, BinLogInfo>();
                }
                // overwrite for latest bin-log info.
                _electionList[key] = logInfo;
            }
        }

        /// <summary>
        /// Compares the log position and id and of mysql instances
        /// and sorts them to create a daisy chain
        /// </summary>
        /// <returns></returns>
        private List<string> BuildDaisyChain()
        {
            List<BinLogInfo> binLogs = null;
            lock(_masterLock)
            {
                binLogs = new List<BinLogInfo>(_electionList.Values);
            }
            binLogs.Sort(new BinLogInfoSorter());
            List<string> nodes = new List<string>();
            foreach (BinLogInfo info in binLogs)
            {
                nodes.Add(info.HostInstanceId);
            }
            return nodes;
        }

        /// <summary>
        /// Declare the winner of master election and broadcasts 
        /// the victory
        /// </summary>
        private void DeclareVictory()
        {
            List<string> chain = BuildDaisyChain();
            string electionResult = string.Empty;
            if (chain == null)
            {
                electionResult = "Election generates no winner!!";                
                return;
            }
            electionResult = "Declare victory with vote count:" + chain.Count;
            OnNewMaster(chain);
            foreach (string instanceID in _electionServices.Keys)
            {
                // use thread to do parallel broadcasting
                BroadcastVictoryThread(instanceID, chain);
            }
        }

        private void BroadcastVictoryThread(string instanceID, List<string> chain)
        {
            Thread sender = new Thread(delegate()
            {
                MasterElectionServiceClient client = _electionServices[instanceID]();
                string msg = "BroadcastVictory to " + instanceID;                
                bool isSuccess = false;
                while (!isSuccess)
                {
                    try
                    {
                        client.BroadcastVictory(chain);
                        isSuccess = true;
                    }
                    catch (Exception)
                    {
                        msg = "Failed: BroadcastVictory to " + instanceID;
                        Thread.Sleep(TimeWait);
                    }
                }
            });
            sender.Start();
        }     

        /// <summary>
        /// Gets fired on election of new master
        /// </summary>
        /// <param name="chain"></param>
        private void OnNewMaster(List<string> chain)
        {
            string masterId = chain[chain.Count - 1];
            if (ValidateMaster() && Master.RoleInstance.Id == masterId)
            {
                // master is unchanged, do nothing
                return;
            }
            _cachedMaster = _instanceIDToInstance[masterId];
            if (NewMaster != null)
            {
                MasterSelectionEventArgs e = new MasterSelectionEventArgs()
                {
                    DaisyChain = chain,
                    MasterId = masterId
                };
                NewMaster(this, e);
            }
        }

        /// <summary>
        /// Holds the current master mysql data
        /// </summary>
        public RoleInstanceEndpoint Master { get { return _cachedMaster; } }

        public bool IsMaster
        {
            get
            {
                if (Master == null)
                {                    
                    return false;
                }                    
                return Master.RoleInstance.Id == _thisInstance.Id;
            }
        }

        public int HostServerId
        {
            get { return _serverId; }
        }

        public string HostInstanceId
        {
            get { return _thisInstance.Id; }
        }

        /// <summary>
        /// Gets the mysql endpoint of the given mysql instance id
        /// </summary>
        /// <param name="instanceId"></param>
        /// <returns></returns>
        public KeyValuePair<string, int> GetMySqlEndpointInfo(string instanceId)
        {
            //TODO: cache MySQL endpoint information
            Role mySqlRole = RoleEnvironment.Roles[MySQLRoleName];
                        
            foreach (RoleInstance endpoint in mySqlRole.Instances)
            {
                if (endpoint.Id == instanceId)
                {
                    return new KeyValuePair<string, int>(
                        endpoint.InstanceEndpoints[MySQLEndpoint].IPEndpoint.Address.ToString(),
                        endpoint.InstanceEndpoints[MySQLEndpoint].IPEndpoint.Port);
                }
            }
            
            return new KeyValuePair<string, int>(string.Empty, 0);
        }

        //NewMaster is fired if a new master is selected
        public event EventHandler<MasterSelectionEventArgs> NewMaster;

        #region IMasterElectionService Members
        // response to other node's proposal by CastElection
        public void ProposeElection(string proposalId)
        {
            if (_electionServices.ContainsKey(proposalId))
            {
                BinLogInfo info = _sqlMonitor.GetBinLogInfo();
                info.HostInstanceId = _thisInstance.Id;
                MasterElectionServiceClient client = _electionServices[proposalId]();
                client.CastElection(_thisInstance.Id, info.LogId, info.Position);
            }
        }

        // the other nodes will response to my proposal
        public void CastElection(string roleInstanceId, int logNumber, long logPosition)
        {
            BinLogInfo info = new BinLogInfo();
            info.LogId = logNumber;
            info.Position = logPosition;
            info.HostInstanceId = roleInstanceId;
            AddElectionListEntry(roleInstanceId, info);
            string voteMsg = string.Format("{0} vote with {1}:{2}", 
                roleInstanceId, logNumber, logPosition);            
            lock (_masterLock)
            {
                if (_electionList.Count == _serverCount) //_electionServices.Count
                {
                    _electionFinishedEvent.Set();

                }
                // the else condition will timeout
                // and election finish with those casted vote
            }
        }

        /// <summary>
        /// Broadcasts the election of the new master
        /// </summary>
        /// <param name="chain"></param>
        public void BroadcastVictory(List<string> chain)
        {
            string masterId = chain[chain.Count - 1];
            if (_runningElection)
            {
                // situation of muliple nodes initiating MasterElection
                // simutaneously
                if (Master == null)
                {
                    // this election is still in progress
                    // ignore incoming result
                    return;
                }
                if (masterId != Master.RoleInstance.Id)
                {
                    // TODO:investigate how the inconsisency arrised!
                    return;
                }
                //OK, TODO: verify if two DaisyChain are the same
                return;
            }
            OnNewMaster(chain);
        }

        /// <summary>
        /// checks if the mysql instance is running
        /// </summary>
        /// <returns></returns>
        public bool IsRunning()
        {
            return MySQLClient.IsRunning(MySQLEndpoint);
        }

        public string QueryMaster()
        {
            if (_sqlMonitor.IsMaster())
            {
                return _thisInstance.Id;
            }
            return null;
        }
        #endregion

        /// <summary>
        /// Log messages 
        /// </summary>
        /// <param name="fmt"></param>
        /// <param name="args"></param>
        private void LogMsg(string fmt, params object[] args)
        {
            _logger.Log(fmt, args);
        }
    }
}
